package cn._51doit.flink.day07;

import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

public class NonKeyedProcessFunctionDemo {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //{"name":"laozhao", "age":18, "fv":9999.99}
        //{"name":"laoduan", "age":28, "fv":999.99}
        //{"name":"laoduan", "age":28, "fv":999.99
        DataStreamSource<String> lines = env.socketTextStream("localhost", 8888);

//        SingleOutputStreamOperator<Boy> boyStream = lines.map(new MapFunction<String, Boy>() {
//            @Override
//            public Boy map(String e) throws Exception {
//                Boy boy = null;
//                try {
//                    boy = JSON.parseObject(e, Boy.class);
//                } catch (Exception ex) {
//                    //ex.printStackTrace();
//                }
//                return boy;
//            }
//        });
//
//        SingleOutputStreamOperator<Boy> filtered = boyStream.filter(boy -> boy != null);


        SingleOutputStreamOperator<Boy> filtered = lines.process(new ProcessFunction<String, Boy>() {

            @Override
            public void processElement(String value, Context ctx, Collector<Boy> out) throws Exception {
                try {
                    Boy boy = JSON.parseObject(value, Boy.class);
                    out.collect(boy);
                } catch (Exception e) {
                    //e.printStackTrace();
                }
            }
        });

        filtered.print();

        env.execute();


    }

    private static class Boy{

        public String name;

        public Integer age;

        public Double fv;

        public Boy(){}

        public Boy(String name, Integer age, Double fv) {
            this.name = name;
            this.age = age;
            this.fv = fv;
        }

        public static Boy of(String name, Integer age, Double fv) {
            return new Boy(name, age, fv);
        }
    }

}
